/*  
 * @(#) KafkaProduct.java Create on 2015年3月11日 下午4:50:50   
 *   
 * Copyright 2015 hiveview.
 */


package com.hiveview.kafka;

import java.util.Enumeration;
import java.util.Properties;
import java.util.ResourceBundle;

import scala.collection.generic.BitOperations.Int;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

/**
 * @KafkaProduct.java
 * @created at 2015年3月11日 下午4:50:50 by zhanghongliang@hiveview.com
 *
 * @desc
 *
 * @author  zhanghongliang@hiveview.com
 * @version $Revision$
 * @update: $Date$
 */
public class KafkaProducer {
	private final kafka.javaapi.producer.Producer<String, String> producer;
	
	public static final ResourceBundle Kafka_producer = ResourceBundle.getBundle("kafka-producer");
	private KafkaProducer() {
		Properties props = new Properties();
		Enumeration<String> enumprops = KafkaProducer.Kafka_producer.getKeys();
		while(enumprops .hasMoreElements()){
			String key = enumprops .nextElement();
			props.put(key, Kafka_producer.getObject(key));
		}
		producer = new kafka.javaapi.producer.Producer<String, String>(
				new ProducerConfig(props ));
	}
	
	private static final class BUILDER{
		private static final KafkaProducer INSTALL = new KafkaProducer();
	}
	
	public static KafkaProducer getInstall(){
		return BUILDER.INSTALL;
	}
	
	public void sendMsg(String topic,String msg){
		this.producer.send(new KeyedMessage<String, String>(topic,(msg.hashCode()/2)+"",msg));
	}
}
